package com.zxinlee.m1;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * 延时消息的使用场景：
 * <p>
 * 比如电商里，提交了一个订单就可以发送一个延时消息，1h后去检查这个订单的状态，如果还是未付款就取消订单释放库存。
 */
public class Consumer {

    public static void main(String[] args) throws MQClientException {
        //创建消费者
        DefaultMQPushConsumer consumer01 = new DefaultMQPushConsumer("consumer01");

        //设置 name server
        consumer01.setNamesrvAddr("192.168.64.141:9876");

        //设置从哪订阅消息
        /*
            标签:
                  Tag01
                  Tag01 || Tag02 || Tag03  接收多个标签
                  *  全部接收
         */
        consumer01.subscribe("Topic01", "Tag01");

        //处理消息的监听器
        //Concurrently 会启动多个线程,并发处理消息
        consumer01.setMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            for (MessageExt msg : msgs) {
                System.out.println("收到消息: " + new String(msg.getBody()));
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });

        //启动消费者
        consumer01.start();

    }
}
